package org.acme.service;

import io.debezium.config.Configuration;
import io.debezium.engine.DebeziumEngine;
import org.acme.message.ChangeData;
import org.acme.utils.JSONUtils;
import io.debezium.engine.format.Json;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Service;

import java.util.Map;
import java.util.Objects;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;


@Service
public class TestService implements CommandLineRunner {

    Logger logger = LoggerFactory.getLogger(getClass());

    @Value("${spring.datasource.server.name}")
    private String servername;

    @Value("${spring.datasource.username}")
    private String username;


    @Value("${spring.datasource.password}")
    private String password;

    @Value("${spring.datasource.db}")
    private String db;


    @Value("${spring.datasource.host}")
    private String host;


    @Value("${spring.datasource.port}")
    private int port;


    @Value("${spring.datasource.serverTimezone}")
    private String serverTimezone;


    @Value("${server.port}")
    private int serverId;

    @Override
    public void run(String... args) throws Exception {
        Executor executor = Executors.newSingleThreadExecutor();
        Configuration config = Configuration.create()
                .with("connector.class", "io.debezium.connector.mysql.MySqlConnector")
                .with("offset.storage", "org.apache.kafka.connect.storage.FileOffsetBackingStore")
                .with("offset.storage.file.filename", "/tmp/offset.dat")
                .with("offset.flush.interval.ms", 60000)
                .with("name", "my-sql-connector")
                .with("database.hostname", host)
                .with("database.port", port)
                .with("database.user", username)
                .with("database.password", password)
                .with("database.connectionTimeZone", serverTimezone)
                .with("database.server.id", serverId)
                .with("database.include.list", db)
                //need listener table
                //.with("table.include.list", db)
                .with("snapshot.mode", "initial")
                .with("database.server.name", servername)
                .with("database.history", "io.debezium.relational.history.FileDatabaseHistory")
                .with("database.history.file.filename", "/tmp/history.dat")
                .build();

        DebeziumEngine engine = DebeziumEngine.create(Json.class)
                .using(config.asProperties())
                .notifying(record -> {
                    receiveChangeEvent(record.value());
                }).build();

        executor.execute(engine);
    }

    private void receiveChangeEvent(String value) {
        if (Objects.nonNull(value)) {
            Map<String, Object> payload = JSONUtils.getPayload(value);
            String handleType = JSONUtils.getHandleType(payload);
            if (!("NONE".equals(handleType) || "READ".equals(handleType))) {
                ChangeData changeData = JSONUtils.getChangeData(payload);
                Map<String, Object> data;
                if ("DELETE".equals(handleType)) {
                    data = changeData.getBefore();
                } else {
                    data = changeData.getAfter();
                }
                org.acme.message.Message message = new org.acme.message.Message();
                message.setData(data);
                message.setDbType("MYSQL");
                message.setDatabase(String.valueOf(changeData.getSource().get("db")));
                message.setTable(String.valueOf(changeData.getSource().get("table")));
                message.setHandleType(handleType);
                logger.info("【Debezium】" + message.toString());
            }
        }
    }


}
